package movee.manager.core;

import movee.config.AppProperties;
import movee.cluster.ClusterService;
import movee.manager.rsync.RsyncClient;
import movee.utils.DelayUtils;
import movee.utils.TimeUtils;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.io.File;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Objects;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * 将文件同步到集群其他节点
 *
 * @author movee
 */
@Service
@Slf4j
public class AppClusterFileSyncer implements Runnable {

    private final AppProperties appProperties;
    private final ClusterService clusterService;
    private final RsyncClient rsyncClient;

    private final BlockingQueue<AppTaskInfo> queue;
    private boolean stop = false;


    public AppClusterFileSyncer(@Qualifier(value = "appScheduledTaskExecutor")
                                           ThreadPoolTaskExecutor clusterFileSyncerExecutor,
                                AppProperties appProperties,
                                ClusterService clusterService,
                                RsyncClient rsyncClient) {
        this.appProperties = appProperties;
        this.clusterService = clusterService;
        this.rsyncClient = rsyncClient;

        this.queue = new ArrayBlockingQueue<>(10000);

        // 启动
        clusterFileSyncerExecutor.execute(this);
    }

    public boolean submit(AppTaskInfo info) {
        if (Objects.nonNull(info)) {
            return queue.offer(info);
        }
        return false;
    }

    @Override
    public void run() {
        while (!stop) {
            try {
                // 阻塞获取任务
                AppTaskInfo info = queue.take();
                List<AppTaskInfo> infos = Lists.newArrayList(info);

                // 读空队列中的所有任务
                int size = queue.size();
                for (int i = 0; i < size; i++) {
                    infos.add(queue.poll());
                }
                // 当前存活的集群peer节点
                List<String> peers = clusterService.getAliveMembers();
                peers.remove(clusterService.getMemberId());

                // 同步目录
                List<String> dirs = new ArrayList<>();
                log.info("sync file to peers. dirs: {}", dirs);
                String currDate = TimeUtils.getStringDateFromDate(LocalDate.now());
                String dst = currDate + File.separator;
                for (String peer : peers) {
                    for (String idc : dirs) {
                        String src = appProperties.getLatestCfgDir() + File.separator + currDate
                                + File.separator + idc.toLowerCase(Locale.ROOT);
                        String host = peer.contains(":") ? peer.substring(0, peer.indexOf(":")) : peer;
                        rsyncClient.syncToRemoteDaemon(host, src, dst);
                    }
                }

                // 延迟1秒
                DelayUtils.delayBySleep(1000L);
            } catch (InterruptedException t) {
                stop = true;
                log.info("exit for InterruptedException");
            } catch (Throwable t) {
                log.error("cluster file sync failed. exception info: {}", ExceptionUtils.getStackTrace(t));
            }
        }
    }


}
